package com.cnebula.dataprocess.sqoop.job;

import java.util.Map;
import java.util.UUID;

import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MToConfig;

import com.cnebula.dataprocess.sqoop.job.adapter.SqoopJobConfigAdapter;
import com.cnebula.dataprocess.sqoop.job.model.ToSqoopJobConstants;

public class HDFSJobConfigAdapter extends SqoopJobConfigAdapter {
	
	private static HDFSJobConfigAdapter instance = null;
	
	static {
		instance = new HDFSJobConfigAdapter();
	}
	
	public static HDFSJobConfigAdapter getInstance(){
		return instance;
	}
	
	private HDFSJobConfigAdapter(){}

	@Override
	public void setMFromConfig(MJob job, Map<String,String> jobConfig) {
		
	}

	@Override
	public void setMToConfig(MJob job, Map<String,String> jobConfig) {
		
		MToConfig toJobConfig = job.getToJobConfig();
		String jobDir = jobConfig.get(ToSqoopJobConstants.HDFS_OUTPUTDIRECTORY);
		if(!jobDir.endsWith("/")){
			jobDir = jobDir + "/";
		}
		String outputDirectory = jobDir + UUID.randomUUID();
        toJobConfig.getStringInput("toJobConfig.outputDirectory").setValue(outputDirectory);
        toJobConfig.getEnumInput("toJobConfig.outputFormat").setValue("TEXT_FILE");
        toJobConfig.getEnumInput("toJobConfig.compression").setValue("NONE");
        toJobConfig.getBooleanInput("toJobConfig.overrideNullValue").setValue(true);
        
	}

}
